package com.marveal.rabbit.producer.broker;

import com.marveal.rabbit.api.Message;
import com.marveal.rabbit.api.MessageType;
import com.marveal.rabbit.producer.constant.BrokerMessageConst;
import com.marveal.rabbit.producer.constant.BrokerMessageStatus;
import com.marveal.rabbit.producer.entity.BrokerMessage;
import com.marveal.rabbit.producer.queue.AsyncBaseQueue;
import com.marveal.rabbit.producer.queue.MessageHolderAsyncQueue;
import com.marveal.rabbit.producer.service.MessageStoreService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;

import java.util.Date;
import java.util.List;

/**
 * $RabbitBrokerImpl 消息实现类
 */
@Slf4j
public class RabbitBrokerImpl implements RabbitBroker {

    private final RabbitTemplateContainer rabbitTemplateContainer;

    private final MessageStoreService messageStoreService;

    public RabbitBrokerImpl(RabbitTemplateContainer rabbitTemplateContainer, MessageStoreService messageStoreService) {
        this.rabbitTemplateContainer = rabbitTemplateContainer;
        this.messageStoreService = messageStoreService;
    }

    @Override
    public void rapidSend(Message message) {
        message.setMessageType(MessageType.RAPID);
        asyncBaseQueueSend(message);
    }

    @Override
    public void confirmSend(Message message) {
        message.setMessageType(MessageType.CONFIRM);
        asyncBaseQueueSend(message);
    }

    @Override
    public void reliantSend(Message message) {
        message.setMessageType(MessageType.RELIANT);
        BrokerMessage msgInDb = messageStoreService.selectByMessageId(message.getMessageId());
        if (msgInDb == null) {
            // 1. 数据库中记录消息发送的日志
            Date now = new Date();
            BrokerMessage brokerMessage = new BrokerMessage()
                    .setMessageId(message.getMessageId())
                    .setMessage(message)
                    .setStatus(BrokerMessageStatus.SENDING.getStatus())
                    .setNextRetry(new Date(now.getTime() + BrokerMessageConst.TIMEOUT))
                    .setCreateTime(now)
                    .setUpdateTime(now);
            messageStoreService.insert(brokerMessage);
        }
        // 2. 执行发送消息
        asyncBaseQueueSend(message);
    }

    @Override
    public void sendMessages() {
        List<Message> messages = MessageHolder.clear();
        messages.forEach(this::messageHolderAsyncQueueSend);
    }

    /**
     * 使用 AsyncBaseQueue线程池 发送消息
     * @param message 消息
     */
    private void asyncBaseQueueSend(Message message) {
        AsyncBaseQueue.submit(() -> {
            doSend(message);
            log.info("#RabbitBrokerImpl.asyncBaseQueueSend# send to rabbitmq, message: {}", message.toString());
        });
    }

    /**
     * 使用 MessageHolderAsyncQueue线程池 发送消息
     * @param message 消息
     */
    private void messageHolderAsyncQueueSend(Message message) {
        MessageHolderAsyncQueue.submit(() -> {
            doSend(message);
            log.info("#RabbitBrokerImpl.messageHolderAsyncQueueSend# send to rabbitmq, message: {}", message.toString());
        });
    }

    /**
     * 发送消息的具体实现
     * @param message 消息
     */
    protected void doSend(Message message) {
        String topic = message.getTopic();
        String routingKey = message.getRoutingKey();
        CorrelationData correlationData = new CorrelationData(String.format("%s#%s#%s",
                message.getMessageId(), System.currentTimeMillis(), message.getMessageType().type));

        RabbitTemplate rabbitTemplate = rabbitTemplateContainer.getRabbitTemplate(message);
        rabbitTemplate.convertAndSend(topic, routingKey, message, correlationData);
    }
}
